
#include "RtBase.h"
#include "RtThreadTask.h"
#include "RtTimerQueueOrderedList.h"

//////////////////////////////////////////////////////////////////////
// class CRtThreadTaskWithEventQueueOnly
//////////////////////////////////////////////////////////////////////

CRtThreadTaskWithEventQueueOnly::CRtThreadTaskWithEventQueueOnly()
{
}

CRtThreadTaskWithEventQueueOnly::~CRtThreadTaskWithEventQueueOnly()
{
}

void CRtThreadTaskWithEventQueueOnly::OnThreadInit()
{
	m_EventQueue.Reset2CurrentThreadId();

	CRtStopFlag::m_Est.Reset2CurrentThreadId();
	SetStartFlag();
}

void CRtThreadTaskWithEventQueueOnly::OnThreadRun()
{
	RT_INFO_TRACE("CRtThreadTaskWithEventQueueOnly::OnThreadRun, Begin.");

	while (!IsFlagStopped()) {
		CRtEventQueueBase::EventsType listEvents;
		RtResult rv = m_EventQueue.PopOrWaitPendingEvents(listEvents, NULL, (DWORD)-1);

		if (RT_SUCCEEDED(rv))
			m_EventQueue.ProcessEvents(listEvents);
	}

	m_EventQueue.DestoryPendingEvents();
	RT_INFO_TRACE("CRtThreadTaskWithEventQueueOnly::OnThreadRun, End.");
}

IRtEventQueue* CRtThreadTaskWithEventQueueOnly::GetEventQueue()
{
	return &m_EventQueue;
}

RtResult CRtThreadTaskWithEventQueueOnly::Stop(CRtTimeValue* aTimeout)
{
	RT_INFO_TRACE("CRtThreadTaskWithEventQueueOnly::Stop");

	RtResult rv = CRtEventStopT<CRtThreadTaskWithEventQueueOnly>::PostStopEvent(this);

	// stop event queue after post stop event.
	m_EventQueue.Stop();
	m_stopped = TRUE;
	return rv;
}


//////////////////////////////////////////////////////////////////////
// class CRtThreadTask
//////////////////////////////////////////////////////////////////////

CRtThreadTask::CRtThreadTask()
	: m_pTimerQueue(NULL)
{
}

CRtThreadTask::~CRtThreadTask()
{
	delete m_pTimerQueue;
}

void CRtThreadTask::OnThreadInit()
{
	// have to new timerqueue in the task thread.
	RT_ASSERTE(!m_pTimerQueue);
	m_pTimerQueue = new CRtTimerQueueOrderedList(NULL);
	RT_ASSERTE(m_pTimerQueue);

	CRtThreadTaskWithEventQueueOnly::OnThreadInit();
}

void CRtThreadTask::OnThreadRun()
{
	RT_INFO_TRACE("CRtThreadTask::OnThreadRun, Begin.");

	while (!IsFlagStopped()) {
		// improve the performance.

		CRtTimeValue tvTimeout(CRtTimeValue::s_tvMax);
		if (m_pTimerQueue) {
			// process timer prior to wait event.
			m_pTimerQueue->CheckExpire(&tvTimeout);
		}

		CRtTimeValue *pTvPara;
		if (tvTimeout == CRtTimeValue::s_tvMax)
			pTvPara = NULL;
		else
			pTvPara = &tvTimeout;

		CRtEventQueueBase::EventsType listEvents;
		RtResult rv = m_EventQueue.PopOrWaitPendingEvents(listEvents, pTvPara);

		// CheckExpire after Wait.
//		m_pTimerQueue->CheckExpire();

		if (RT_SUCCEEDED(rv))
			m_EventQueue.ProcessEvents(listEvents);
	}

	m_EventQueue.DestoryPendingEvents();
	RT_INFO_TRACE("CRtThreadTask::OnThreadRun, End.");
}

IRtTimerQueue* CRtThreadTask::GetTimerQueue()
{
	return m_pTimerQueue;
}

//////////////////////////////////////////////////////////////////////
// class CRtEventQueueUsingConditionVariable
//////////////////////////////////////////////////////////////////////

CRtEventQueueUsingConditionVariable::CRtEventQueueUsingConditionVariable()
	: m_Condition(m_Mutex)
{
}

CRtEventQueueUsingConditionVariable::~CRtEventQueueUsingConditionVariable()
{
}

RtResult CRtEventQueueUsingConditionVariable::
PostEvent(IRtEvent *aEvent, EPriority aPri)
{
	// Don't hold the mutex when signaling the condition variable.
	{
		CRtMutexGuardT<MutexType> theGuard(m_Mutex);
		RtResult rv = CRtEventQueueBase::PostEvent(aEvent, aPri);
		if (RT_FAILED(rv))
			return rv;
	}

	// Don't care the error if Signal() failed.
	m_Condition.Signal();
	return RT_OK;
}

RtResult CRtEventQueueUsingConditionVariable::
PopOrWaitPendingEvents(CRtEventQueueBase::EventsType &aEvents, 
					   CRtTimeValue *aTimeout, DWORD aMaxCount)
{
	RtResult rv;
	CRtMutexGuardT<MutexType> theGuard(m_Mutex);

	if (m_Events.empty()) {
		rv = m_Condition.Wait(aTimeout);
		if (RT_FAILED(rv) && rv != RT_ERROR_TIMEOUT) {
			RT_ERROR_TRACE("CRtEventQueueUsingConditionVariable::PopOrWaitPendingEvents,"
				"m_Events is not empty. nSize=" << m_dwSize << " rv=" << rv);
		//	return rv;
		}
	}
	
	return PopPendingEvents(aEvents, aMaxCount);
}


//////////////////////////////////////////////////////////////////////
// class CRtThreadHeartBeat
//////////////////////////////////////////////////////////////////////

CRtThreadHeartBeat::CRtThreadHeartBeat()
	: m_pTimerQueue(NULL)
{
}

CRtThreadHeartBeat::~CRtThreadHeartBeat()
{
	delete m_pTimerQueue;
}

void CRtThreadHeartBeat::OnThreadInit()
{
	// have to new timerqueue in the task thread.
	RT_ASSERTE(!m_pTimerQueue);
	m_pTimerQueue = new CRtTimerQueueOrderedList(NULL);
	RT_ASSERTE(m_pTimerQueue);

	m_EventQueue.Reset2CurrentThreadId();
}

IRtEventQueue* CRtThreadHeartBeat::GetEventQueue()
{
	return &m_EventQueue;
}

IRtTimerQueue* CRtThreadHeartBeat::GetTimerQueue()
{
	return m_pTimerQueue;
}

void CRtThreadHeartBeat::OnThreadRun()
{
	DoHeartBeat();
}

RtResult CRtThreadHeartBeat::Stop(CRtTimeValue* aTimeout)
{
	return RT_OK;
}

RtResult CRtThreadHeartBeat::DoHeartBeat()
{
	if (m_pTimerQueue)
		m_pTimerQueue->CheckExpire();

	CRtEventQueueBase::EventsType listEvents;
	RtResult rv = m_EventQueue.PopPendingEventsWithoutWait(listEvents, (DWORD)-1);
	if (RT_SUCCEEDED(rv))
		m_EventQueue.ProcessEvents(listEvents);

	return RT_OK;
}
